#include <boost/bind.hpp>

#include "buffers.h"
#include "socket_io.h"
#include "dhd_fs.pb.h"


namespace dhd
{

   using boost::asio::detail::socket_ops::host_to_network_short;
   using boost::asio::detail::socket_ops::network_to_host_short;
   
   
   SocketIO::SocketIO( tcp::socket * _sock, 
                       ssl_socket  * _ssl_sock,
                       MessageArrivedCallback _cb,
                       IOErrorCallback _ecb,
                       bool _delete_sockets ) :
       sock(_sock), ssl_sock(_ssl_sock), disconnected(false),
       transmit_pending( false ), on_arrive_callback(_cb),
       io_error_callback(_ecb), delete_sockets(_delete_sockets)
       {
           queued_buffs = new BufferVector;
           tx_buffs     = new BufferVector;
           
           queued_lens  = new LenVector;
           tx_lens      = new LenVector;
           
           queued_vec   = new TransmitVector;
           tx_vec       = new TransmitVector;
           
           async_read_next();
       }
   
   
   SocketIO::~SocketIO()
   {
       delete queued_buffs;
       delete tx_buffs;
       delete queued_lens;
       delete tx_lens;
       delete queued_vec;
       delete tx_vec;
       
       if (delete_sockets)
       {
           delete sock;
           delete ssl_sock;
       }
       
       while ( !len_stack.empty() )
       {
           delete len_stack.top();
           len_stack.pop();
       }
   }
   
   
   void SocketIO::report_error( const char * msg, bool reconnect )
   {
       disconnected = true;
       io_error_callback( msg, reconnect );
   }
   
   
   void SocketIO::enqueue( uint16_t request_id, PBuffer header, PBuffer buffer )
   {
       boost::mutex::scoped_lock( q_mutex );
       
       if ( len_stack.empty() )
           len_stack.push( new TxLens );
       
       TxLens * lens = len_stack.top();
       len_stack.pop();
       
       lens->header_len = host_to_network_short( header->size() );
       lens->buffer_len = host_to_network_short( buffer ? buffer->size() : 0 );
       lens->request_id = host_to_network_short( request_id );
       
       queued_lens->push_back( lens );
       queued_buffs->push_back( header );
       
       queued_vec->push_back( boost::asio::buffer( lens, sizeof(*lens) ) );
       queued_vec->push_back( boost::asio::buffer( header->data(), header->size() ) );
       
       if (buffer && buffer->size() != 0)
       {
           queued_buffs->push_back( buffer );
           queued_vec->push_back( boost::asio::buffer( buffer->data(), buffer->size() ) );
       }
       
       if ( ! transmit_pending )
           begin_transmit();
   }
   
   
   
   void SocketIO::begin_transmit()
   {   
       if (disconnected) return;
   
       // Note: This method is called with the mutex lock held
       
       //------------------------------------
       // Swap the Queued & Transmit buffers
       BufferVector   * tmp_buf = queued_buffs;
       LenVector      * tmp_len = queued_lens;
       TransmitVector * tmp_vec = queued_vec;
       
       queued_buffs = tx_buffs;
       tx_buffs     = tmp_buf;
       
       queued_lens  = tx_lens;
       tx_lens      = tmp_len;
       
       queued_vec   = tx_vec;
       tx_vec       = tmp_vec;
       //------------------------------------
       
       transmit_pending = true;
       
       if (ssl_sock)
       {
           boost::asio::async_write( *ssl_sock, 
                                     *tx_vec,
                                     boost::bind( &SocketIO::on_async_write_complete,
                                                  this, 
                                                  boost::asio::placeholders::error,
                                                  boost::asio::placeholders::bytes_transferred,
                                                  PSocketIO(this)));
       }
       else
       {
           boost::asio::async_write( *sock, 
                                     *tx_vec,
                                     boost::bind( &SocketIO::on_async_write_complete,
                                                  this, 
                                                  boost::asio::placeholders::error,
                                                  boost::asio::placeholders::bytes_transferred,
                                                  PSocketIO(this)));
       }
       
       
   }
   
   
   void SocketIO::on_async_write_complete( const boost::system::error_code& error, std::size_t bytes_transferred, boost::intrusive_ptr<SocketIO> ref )
   {
       if (disconnected) return;
       
       if (error)
       {
           report_error("Write failed", true);
           return;
       }
       
       boost::mutex::scoped_lock( q_mutex );
       
       transmit_pending = false;
       
       for( LenVector::iterator i = tx_lens->begin(); i != tx_lens->end(); i++ )
           len_stack.push( *i );
       
       tx_vec->clear();
       tx_lens->clear();
       tx_buffs->clear();
       
       if ( ! queued_vec->empty() )
           begin_transmit();
   }
   
   
   //-----------------------------------------------------------------------------------------
   // Read
   //
   void SocketIO::async_read_next()
   {
       if (disconnected) return;
       
       if (ssl_sock)
       {
           boost::asio::async_read( *ssl_sock,
                                    boost::asio::buffer( msg_lens ), 
                                    boost::bind( &SocketIO::on_async_read_complete,
                                                 this, 
                                                 boost::asio::placeholders::error,
                                                 boost::asio::placeholders::bytes_transferred,
                                                 PSocketIO(this)));
       }
       else
       {
           boost::asio::async_read( *sock, 
                                    boost::asio::buffer( msg_lens ), 
                                    boost::bind( &SocketIO::on_async_read_complete,
                                                 this, 
                                                 boost::asio::placeholders::error,
                                                 boost::asio::placeholders::bytes_transferred,
                                                 PSocketIO(this)));
       }
   }
   
   
   void SocketIO::on_async_read_complete( const boost::system::error_code& error, std::size_t bytes_transferred, boost::intrusive_ptr<SocketIO> ref )
   {
       if (disconnected) return;
       
       if (error)
       {
           report_error("Read Failed", true);
           return;
       }
       
       uint16_t  header_len = network_to_host_short( msg_lens[0] );
       uint16_t  buff_len   = network_to_host_short( msg_lens[1] );
       uint16_t  request_id = network_to_host_short( msg_lens[2] );
       
       char     *hbuff      = 0;
       char     *dbuff      = 0;
       
       PBuffer   pheader    = allocate_buffer( header_len );
       PBuffer   pbuffer    = 0;
       
       std::size_t    read_len   = 0;
       dhd::ToClient *result     = 0;
       
       
       if (header_len > MAX_MSG_LEN || buff_len > MAX_MSG_LEN)
       {
           report_error("Requested read exceeds maximum length", false);
           return;
       }
       
       // ------------- Synchronous Read Reply -----------
       // 
       //  
       if (buff_len)
       {   
           pbuffer = allocate_buffer( buff_len );
           
           boost::array<boost::asio::mutable_buffer, 2> arrs = { boost::asio::buffer( pheader->data(), header_len ),
               boost::asio::buffer( pbuffer->data(), buff_len )
           };
           if (ssl_sock)
               read_len = boost::asio::read( *ssl_sock, arrs );
           else
               read_len = boost::asio::read( *sock, arrs );
       }
       else
       {
           if (ssl_sock)
               read_len = boost::asio::read( *ssl_sock, boost::asio::buffer( pheader->data(), header_len ) );
           else
               read_len = boost::asio::read( *sock, boost::asio::buffer( pheader->data(), header_len ) );
       }
       
       if (read_len != buff_len + header_len)
       {
           report_error("Failed to read reply message correctly", true);
           return;
       }

       try
       {
           on_arrive_callback( request_id, pheader, pbuffer );
       
           async_read_next();
       }
       catch( BadProtocol & e )
       {
           report_error( e.what(), e.reconnect );
       }
   }
   
}// end namespace dhd